package net.bwie.vehicle.dws.job1;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import net.bwie.vehicle.dws.bean1.AvgBatteryLevelResult;
import net.bwie.vehicle.dws.bean1.HeatmapDataResult;
import net.bwie.vehicle.dws.bean1.OnlineCountResult;
import net.bwie.vehicle.dws.bean1.VehicleData;
import net.bwie.vehicle.dws.function1.BatteryLevelAggregatorFunction;
import net.bwie.vehicle.dws.function1.HeatmapCounterFunction;
import net.bwie.vehicle.dws.function1.OnlineVehicleCounterFunction;
import net.bwie.vehicle.dws.utils.GeoHashUdf;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
/*
   实时看板
 */
public class RealTimeSpectacularsJob {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(1);
        // Kafka数据源
        DataStream<String> stringDataStream = KafkaUtil.consumerKafka(env, "car-vehicle-data");
        SingleOutputStreamOperator<VehicleData> vehicleStream = stringDataStream.map(json ->
                JSON.parseObject(json, VehicleData.class));

        // 分配时间戳和水位线
        DataStream<VehicleData> timedStream = vehicleStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                                .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
                )
                .name("assign-timestamps");

        // 1. 实时在线车辆数统计 (每30秒滚动窗口)
        DataStream<OnlineCountResult> onlineVehicles = timedStream
                .keyBy(data -> "all")
                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
                .process(new OnlineVehicleCounterFunction())
                .map(count -> new OnlineCountResult("实时在线车辆数", count))
                .name("online-vehicles-counter");
        // 添加CK库
        SingleOutputStreamOperator<String> map1 = onlineVehicles.map(new MapFunction<OnlineCountResult, String>() {
            @Override
            public String map(OnlineCountResult value) throws Exception {
                String  output = value.getName() + ',' + value.getCount();
                return output;
            }
        });
        String upsertSql = "insert into new_car.dws_onlineVehicles (name, countCar)\n" +
                "values (?, ?)";
        JdbcUtil.sinkToClickhouseUpsert(map1, upsertSql);

        // 2. 平均电池健康度统计 (每5分钟滚动窗口)
        DataStream<AvgBatteryLevelResult> avgBatteryLevel = timedStream
                .keyBy(data -> "all")
                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
                .aggregate(new BatteryLevelAggregatorFunction())
                .map(avg -> new AvgBatteryLevelResult("平均电池健康度", avg))
                .name("avg-battery-level");
        // 添加CK库
        SingleOutputStreamOperator<String> map2 = avgBatteryLevel.map(new MapFunction<AvgBatteryLevelResult, String>() {
            @Override
            public String map(AvgBatteryLevelResult value) throws Exception {
                String output = value.getName() + ',' + value.getAvgLevel();
                return output;
            }
        });
        String upsertSql1 = "insert into new_car.dws_avgBatteryLevel (name, avgcell)\n" +
                "values (?, ?)";
        JdbcUtil.sinkToClickhouseUpsert(map2, upsertSql1);

        // 3. 车辆地理分布热力图 (输出原始经纬度数据)30秒的滚动窗
        DataStream<HeatmapDataResult> heatmapStream = timedStream
                .keyBy(data -> GeoHashUdf.getGeoHash(data.getLatitude(), data.getLongitude()))
                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
                .process(new HeatmapCounterFunction())
                .map(map -> {
                    HeatmapDataResult result = null;
                    for (Map.Entry<String, Integer> entry : map.entrySet()) {
                        result = new HeatmapDataResult(entry.getKey(), entry.getValue());
                    }
                    return result;
                });
        SingleOutputStreamOperator<String> map3 = heatmapStream.map(new MapFunction<HeatmapDataResult, String>() {
            @Override
            public String map(HeatmapDataResult value) throws Exception {
                String output = value.getGeohash() + ',' + value.getCount();
                return output;
            }
        });
        String upsertSql2 = "insert into new_car.dws_heatmapStream (name, countCar)\n" +
                "values (?, ?)";
        JdbcUtil.sinkToClickhouseUpsert(map3, upsertSql2);
        // 输出结果
        onlineVehicles.print("实时在线车辆数统计");
//        avgBatteryLevel.print("平均电池健康度统计");
//        heatmapStream.print("车辆地理分布热力图");

        // 执行程序
        env.execute("RealTimeSpectacularsJob");
    }


    }